package com.bossien.flink.factory;

import com.bossien.common.common.SpringContextHolder;
import com.bossien.flink.config.ExecutionTableConfig;
import com.bossien.flink.config.RMQExcutionStreamHandlerConfig;
import com.bossien.flink.interfaces.IEnviroment;
import com.bossien.flink.interfaces.IExecutionFactory;
import com.bossien.flink.interfaces.IExecutionHandlerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.lang.reflect.Field;

/**
 * @author 胡春林
 * 获取具体的执行器
 */
@Slf4j
@Component
public class ExcutionFactory implements IExecutionFactory {

    @Override
    public <T> T newInstance(Class<T> clazz) {

        try {
            if (null != clazz && clazz.getGenericSuperclass().getClass().isInstance(RMQExcutionStreamHandlerConfig.class)) {
                return clazz.newInstance();
            }
        }catch (IllegalAccessException e)
        {
            e.printStackTrace();
            log.info("获取实例失败，具体原因：{}",e.getCause());
            return null;

        }catch (InstantiationException ex)
        {
            ex.printStackTrace();
            log.info("获取实例失败，具体原因：{}",ex.getCause());
            return null;
        }
        return null;
    }

    /**
     * 获取rabbitmq环境
     * @param clazz
     * @param sourceAndSkin false 只接受mq数据 true 接受mq数据并发送至MQ
     * @return
     */
    @Override
    public boolean getRabbitInstance(Class clazz, boolean sourceAndSkin) {

     try {
         log.info("获取父类类型 进行isInstance()判断");
        if(null == clazz || !(IExecutionHandlerConfig.class.isAssignableFrom(clazz.getSuperclass())))
        {
            log.info("当前运行的任务不属于RMQ消息服务中间件,请确认,警告类：{}",this.getClass().getName());
            throw new Exception("当前运行的任务不属于RMQ消息服务中间件");
        }
        return  initialize(clazz,sourceAndSkin);
        }catch (Exception e)
        {
            e.printStackTrace();
            log.info("任务提交失败，详细原因：{}",e.getMessage());
            return false;
        }
    }

    //开始执行具体任务
    public boolean initialize(Class<? extends IExecutionHandlerConfig> clazz, boolean sourceAndSkin) throws Exception{

            String ExcutionEnvironmentName = "streamExecutionEnvironment";
            //开始运行作业
            IExecutionHandlerConfig  examDataStreamExecute =  SpringContextHolder.getBean(clazz);
            //获取系统环境对象
            if(examDataStreamExecute instanceof ExecutionTableConfig)
            {
                ExcutionEnvironmentName = "tableEnvironment";
            }
            //提交作业
            Field field = null;
            try {
                 field = clazz.getSuperclass().getDeclaredField(ExcutionEnvironmentName);
            }catch (NoSuchFieldException e)
            {
                e.printStackTrace();
                log.info("获取{}系统环境发生异常",ExcutionEnvironmentName);
            }
            if(null == field) {
               throw new Exception("获取系统环境为空，请检查相关实现类");
            }
            field.setAccessible(true);
            //分俩套开始提交
            if(examDataStreamExecute instanceof ExecutionTableConfig)
            {
                //执行过滤器
                Table table = (Table) examDataStreamExecute.executeFilter(examDataStreamExecute.addSource());
                if(sourceAndSkin) {examDataStreamExecute.addSkin(table); }
                //TabelApi层面环境
                StreamTableEnvironment streamTableEnv = (StreamTableEnvironment) field.get(examDataStreamExecute);
                streamTableEnv.execEnv().execute();
                log.info("---------基于Table API层面环境开始执行--------");

            }else {
                //执行过滤器
                DataStream dataStream = (DataStream) examDataStreamExecute.executeFilter(examDataStreamExecute.addSource());
                if(sourceAndSkin) {examDataStreamExecute.addSkin(dataStream); }
                //DataStream流式层面环境
                StreamExecutionEnvironment streamExecutionEnv = (StreamExecutionEnvironment) field.get(examDataStreamExecute);
                streamExecutionEnv.execute();
                log.info("--------基于DataStream API层面环境开始执行-------");
            }
            return true;

    }

}
